/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.util.Comparator;
import java.util.logging.Logger;

import edu.ucsd.db.bassarisk.data.api.IValueType;
import edu.ucsd.db.bassarisk.data.api.IValueType.IObjectValueReference;
import edu.ucsd.db.bassarisk.data.api.IValueType.ITypeTools;
import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.storage.common.IColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.reading.CabinetReader.DrawerBuffer;
import edu.ucsd.db.bassarisk.storage.reading.IColumnIndexReader.DrawerInfo;

public class ColumnReader<VT extends IValueType> implements IColumnSegmentReader<VT> {
  
  private ColumnInfo columnInfo;
  private DrawerBuffer dataDrawer;
  private IDrawerBufferSource drawerCache;
  private final DrawerInfo<VT> drawerInfo;
  private final DrawerReader<VT> drawerReader;
  private DrawerBuffer headerDrawer;
  private final IColumnIndexReader<VT> indexReader;

  protected Comparator<IValueReference<VT>> valueComparator;
  
  private boolean initialized;
  
  private static final Logger LOG = Logger.getLogger(ColumnReader.class.getName());
  
  public ColumnReader(DrawerReader<VT> drawerReader, IColumnIndexReader<VT> indexReader, 
                      ITypeTools<VT> columnTypeTools) {
    this.drawerReader = drawerReader;
    this.indexReader = indexReader;
    this.drawerInfo = new DrawerInfo<VT>(columnTypeTools);
    this.initialized = false;
    // FIXME: This is the slow way to compare, we can use fast non-generic 
    //        implementations once the query engine is working.
    this.valueComparator = columnTypeTools.getGenericComparator();
  }
  
  public void close() {
    this.headerDrawer.close();
    this.dataDrawer.close();
  }
  
  public long currentIndex() {
    return this.drawerReader.currentIndex();
  }
  
  @Override
  public boolean hasEntryIndex(long targetEntryIndex) {
    long firstIndex = columnInfo.getFirstEntryIndex();
    long numEntries = columnInfo.getEntryCount();
    
    return targetEntryIndex >= firstIndex && targetEntryIndex < (firstIndex + numEntries);
  }
  
  @Override
  public boolean hasValue(IValueReference<VT> targetValue) {
    IValueReference<VT> firstValue = this.columnInfo.getFirstValue();
    IValueReference<VT> lastValue = this.columnInfo.getFirstValue();
    
    return this.valueComparator.compare(targetValue, firstValue) >= 0
        && this.valueComparator.compare(targetValue, lastValue) <= 0;
  }
  
  public class ColumnInfo implements IColumnSegmentInfo<VT> {
    private long entryCount;
    private IObjectValueReference<VT> firstValue;
    private IObjectValueReference<VT> lastValue;
    
    public void reset(long entryCount,
                      IObjectValueReference<VT> firstValue,
                      IObjectValueReference<VT> lastValue) {
      this.entryCount = entryCount;
      this.firstValue = firstValue;
      this.lastValue = lastValue;
    }

    @Override
    public long getEntryCount() {
      return this.entryCount;
    }

    @Override
    public long getFirstEntryIndex() {
      return 0;
    }

    @Override
    public IObjectValueReference<VT> getFirstValue() {
      return this.firstValue;
    }

    @Override
    public IObjectValueReference<VT> getLastValue() {
      return this.lastValue;
    }

    @Override
    public boolean hasEntry(long entryIndex) {
      return entryIndex < this.entryCount;
    }
    
  }
  
  public void initialize(IDrawerBufferSource drawerCache, int columnAddress,
                         ITypeTools<VT> columnTypeTools) {
    LOG.info("Initialize column at " + columnAddress);
    int columnIndexDrawer = (columnAddress >> 8) & 0x00FFFFFF;
    int columnIndexHeaderCard = columnAddress & 0xFF;
    
    this.drawerCache = drawerCache;
    this.headerDrawer = this.drawerCache.makeDrawerBuffer();
    this.dataDrawer = this.drawerCache.makeDrawerBuffer();
    
    drawerCache.setDrawer(columnIndexDrawer, this.headerDrawer);
    this.columnInfo = new ColumnInfo();
    this.indexReader.initialize(this.headerDrawer.getBuffer(), this.columnInfo,
                                columnIndexHeaderCard);
    this.initialized = true;
  }
  
  @Override
  public boolean isInitialized() { return this.initialized; }
  
  @Override
  public void jumpCursor(int targetEntryIndex, IValueReference<VT> valueOut) {
    assert valueOut != null;
    
    if (!this.drawerReader.hasEntryIndex(targetEntryIndex)) {
      boolean hasDrawer =
          this.indexReader.setDrawerInfoForEntryIndex(targetEntryIndex, this.drawerInfo);
      assert hasDrawer; // jumpCursor should not be called for invalid indexes
      this.drawerCache.setDrawer(this.drawerInfo.getDrawerIndex(), this.dataDrawer);
      this.drawerReader.resetDrawer(this.dataDrawer.getBuffer(), this.drawerInfo);
    }
    
    this.drawerReader.jumpCursor(targetEntryIndex, valueOut);
  }
  
  @Override
  public boolean jumpCursorToValue(IValueReference<VT> minimumTargetValue,
                                   IValueReference<VT> valueOut) {
    assert valueOut != null;
    
    if (!this.drawerReader.hasValue(minimumTargetValue)) {
      boolean hasDrawer = 
          this.indexReader.setDrawerInfoForValue(minimumTargetValue, this.drawerInfo);
      if (!hasDrawer) {
        return false; // We have exceeded the domain for this column.
      }
      this.drawerCache.setDrawer(this.drawerInfo.getDrawerPageId(), this.dataDrawer);
      this.drawerReader.resetDrawer(this.dataDrawer.getBuffer(),  this.drawerInfo);
    }

    return this.drawerReader.jumpCursorToValue(minimumTargetValue, valueOut);
  }
    
  
  @Override
  public boolean stepCursor(IValueReference<VT> valueOut) {
    assert valueOut != null;
    
    if (this.drawerReader.stepCursor(valueOut)) {
      return true;
    } else if (this.indexReader.setNextDrawerInfo(this.drawerInfo)) {
      this.drawerCache.setDrawer(this.drawerInfo.getDrawerPageId(), this.dataDrawer);
      this.drawerReader.resetDrawer(this.dataDrawer.getBuffer(), this.drawerInfo);
      boolean stepped = this.drawerReader.stepCursor(valueOut);
      assert stepped;
      return true;
    } else {
      return false;
    }
  }
  
}
